package com.example.transform;

import com.example.beans.SensorReading;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Collections;

public class TransformTest4_MultipleStreams {

    //  TypeInformation.of(SensorReading.class) 当前的数据类型
    private static final OutputTag<SensorReading> HIGH = new OutputTag<>("high", TypeInformation.of(SensorReading.class));
    private static final OutputTag<SensorReading> LOW = new OutputTag<>("low", TypeInformation.of(SensorReading.class));


    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 从文件读取数据
        DataStream<String> inputStream = env.readTextFile("E:\\hjx\\ideapro\\myflinkstu\\src\\main\\resources\\sensor.txt");

        // 转换成SensorReading
        DataStream<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        } );

        // 1. 分流，按照温度值30度为界分为两条流
        SingleOutputStreamOperator<SensorReading> process = dataStream.process(new ProcessFunction<SensorReading, SensorReading>() {
            @Override
            public void processElement(SensorReading value, Context ctx, Collector<SensorReading> out) throws Exception {
                if (value.getTemperature() > 30) {
                    ctx.output(HIGH, value);
                } else {
                    ctx.output(LOW, value);
                }
            }
        });

        DataStream<SensorReading> high = process.getSideOutput(HIGH);
        DataStream<SensorReading> low = process.getSideOutput(LOW);

//        high.print("high");
//        low.print("low");

        // 已过时方法 split()

//        SplitStream<SensorReading> splitStream = dataStream.split(new OutputSelector<SensorReading>() {
//            @Override
//            public Iterable<String> select(SensorReading value) {
//                return (value.getTemperature() > 30) ? Collections.singletonList("high") : Collections.singletonList("low");
//            }
//        });
//
//        DataStream<SensorReading> highTempStream = splitStream.select("high");
//        DataStream<SensorReading> lowTempStream = splitStream.select("low");
//        DataStream<SensorReading> allTempStream = splitStream.select("high", "low");
//
//        highTempStream.print("high");
//        lowTempStream.print("low");
//        allTempStream.print("all");
//
        // 2. 合流 connect，将高温流转换成二元组类型，与低温流连接合并之后，输出状态信息
        DataStream<Tuple2<String, Double>> warningStream = high.map(new MapFunction<SensorReading, Tuple2<String, Double>>() {
            @Override
            public Tuple2<String, Double> map(SensorReading value) throws Exception {
                // (senderID,temp) 将高温的 id 和 温度 提取出来
                return new Tuple2<>(value.getId(), value.getTemperature());
            }
        });


        // 得到连接流
        ConnectedStreams<Tuple2<String, Double>, SensorReading> connectedStreams = warningStream.connect(low);

        /*
        CoMapFunction, CoFlatMapFunction
        作用于 ConnectedStreams 上，功能与 map
        和 flatMap 一样，对 ConnectedStreams 中的每一个 Stream 分别进行 map 和 flatMap
        处理。
         */
        DataStream<Object> resultStream = connectedStreams.map(new CoMapFunction<Tuple2<String, Double>, SensorReading, Object>() {
            @Override
            public Object map1(Tuple2<String, Double> value) throws Exception {
                // (senderID,temp,报警信息(状态))
                return new Tuple3<>(value.f0, value.f1, "high temp warning");
            }

            @Override
            public Object map2(SensorReading value) throws Exception {
                // (senderID,报警信息(状态))
                return new Tuple2<>(value.getId(), "normal");
            }
        });

//        resultStream.print();

        // 3. union联合多条流
//        warningStream.union(lowTempStream);
//        high.union(low,low);
//        DataStream<SensorReading> union = high.union(low);
//        union.map(new MapFunction<SensorReading, Object>() {
//            @Override
//            public Object map(SensorReading value) throws Exception {
//                return null;
//            }
//        });

//        union.print("union");

        env.execute();
    }
}
